MediaLiveでの配信終了時に自動でアーカイブファイルを変換する仕組みを作ってみた

MediaLiveでの配信終了時に自動でアーカイブファイルをMediaConvertで変換する仕組みを作ってみました。

#AWS MediaLive

#AWS MediaConvert

#CloudWatch Events

#AWS Lambda

#AWSメディアサービス

大前諒祐

2020.01.27

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、大前です。

MediaLive での配信終了時に自動で MediaConvert の変換ジョブを走らせる仕組みを作ってみました。

構成図

構成は以下の通りです。

MediaLive から S3 にアーカイブを出力するように設定
CloudWatch Events で MediaLive チャネルのステータスを監視
MediaLive チャネルが「停止」になったら指定の Lambda 関数を呼び出す
Lambda 関数から MediaConvert の変換ジョブをリクエスト
MediaConvert がアーカイブファイルに対して変換ジョブを実行

やってみた

実際にやってみます。

1. MediaLive チャネルを作成

まずは、MediaLive の設定をしていきます。チャネル作成時の詳細な手順を確認したい場合はこちらを参照ください。

チャネル名を入力し、ロールを選択

入力をアタッチ。今回は OBS からテストするので RTMP プッシュの入力を作っておきました。

出力にアーカイブグループを追加。実際に運用する時は、この他に配信用の出力を作成する形となります。

アーカイブの送信先となる S3 を指定。

出力設定は以下。

ここまで設定したら、「作成」を押下します。

2. MediaConvert でジョブテンプレートを作成

MediaConvert でジョブテンプレートを作成します。

ジョブテンプレートの作成から、以下のようなテンプレートを作成しました。

入力は Lambda 側で設定する為、テンプレートでは何も設定しないのがポイントです。

出力は、よしなに設定してください。

3. Lambda 関数を作成

続いて、MediaConvert のジョブを呼び出す Lambda 関数を作成します。

作成にあたっては、こちらを参考にしました。

ランタイムは Python 3.7、ロールはとりあえずデフォルトで作成します。

実行ロールに、以下のポリシーをアタッチします。

MediaConvertRole の中身は以下です。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "iam:PassRole",
                "mediaconvert:CreateJob"
            ],
            "Resource": "*"
        }
    ]
}

ソースは以下になります。

import json
import urllib.parse
import boto3
import os

MEDIACONVERT_ENDPOINT = os.environ['MEDIACONVERT_ENDPOINT'] # MediaConvertのエンドポイント
TARGET_BUCKET = os.environ['TARGET_BUCKET'] # アーカイブファイルが格納されているバケット
TARGET_BUCKET_PREFIX = os.environ['TARGET_BUCKET_PREFIX'] # アーカイブファイルのキー（フォルダ名）
JOB_TEMPLATE = os.environ['JOB_TEMPLATE'] # MediaConvertのジョブテンプレート
QUEUE = os.environ['QUEUE'] # MediaConvertのキューARN
ROLE = os.environ['ROLE'] # MediaConvertの実行ロール

def lambda_handler(event, context):
    # クライアント初期化
    s3 = boto3.client('s3')
    mediaconvert =  boto3.client('mediaconvert', region_name='ap-northeast-1', endpoint_url=MEDIACONVERT_ENDPOINT)
    
    # アーカイブファイルのリスト取得
    targetlist = get_targetlist(s3)
    if len(targetlist) == 0:
        # ファイルが見つからなかったら終了
        return
    
    # MediaConvertのリクエスト生成
    input_objects = []
    for target in targetlist:
        with open("input_template.json", "r") as jsonfile:
            input_object = json.load(jsonfile)
        input_object["FileInput"] = "s3://" + TARGET_BUCKET + "/" + target
        input_objects.append(input_object)
    print('input_objects : ' + str(input_objects))
    
    with open("job.json", "r") as jsonfile:
        job_object = json.load(jsonfile)
    job_object["Inputs"] = input_objects
    print('job_object : ' + str(job_object))
    
    # リクエスト実行
    response = mediaconvert.create_job(
      JobTemplate = JOB_TEMPLATE,
      Queue = QUEUE,
      Role = ROLE,
      Settings=job_object
    )
        
    return {
        'statusCode': 200
    }

def get_targetlist(s3):
    # オブジェクト取得
    res = s3.list_objects_v2(
        Bucket = TARGET_BUCKET,
        Prefix = TARGET_BUCKET_PREFIX
        )
    result_list = []

    for obj in res['Contents']:
        # list_objectsでフォルダ名のみの文字列も返却されるため弾く
        if obj['Key'].split('/')[1] != '':
            result_list.append(obj['Key'])
    print('result_list : ' + str(result_list))
    return result_list

処理の概要としては、以下になります。

アーカイブファイルが格納されている S3 バケットから変換対象のファイル一覧を取得
MediaConvert のリクエストを作成
MediaConvert にジョブのリクエストを実行

細かいエラーハンドリング等はしていないので、引用する場合には適宜改修してお使いください。

ソース内で指定しているファイルは、以下になります。

MediaConvert で作成したジョブテンプレートの JSON を取得出来るので、それを元にして動的に入力を設定する為のファイルを作成しました。

input_template.json

{
    "AudioSelectors": {
      "Audio Selector 1": {
        "Offset": 0,
        "DefaultSelection": "DEFAULT",
        "ProgramSelection": 1
      }
    },
    "VideoSelector": {
      "ColorSpace": "FOLLOW",
      "Rotate": "DEGREE_0",
      "AlphaBehavior": "DISCARD"
    },
    "FilterEnable": "AUTO",
    "PsiControl": "USE_PSI",
    "FilterStrength": 0,
    "DeblockFilter": "DISABLED",
    "DenoiseFilter": "DISABLED",
    "TimecodeSource": "EMBEDDED",
    "FileInput": ""
}

job.json